// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: xguo

#include "coin2/exchange/binance_futures/order_v1/parser.h"

#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/binance/order_v1/error_code.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::binance_futures::order_v1 {

namespace {

using namespace coin2::exchange::binance::order_v1::error_code;

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::binance::symbology::BinanceCurrency;
using coin2::exchange::binance_futures::symbology::BinanceFuturesProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using TransferType = coin::proto::TransferRequestProto::TransferType;

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clientOrderId")) {
    std::string proc_str = data["clientOrderId"].GetString();
    try {
      proc_order_id = stoll(proc_str);
    } catch (std::exception& e) {
      LOG(ERROR) << e.what() << proc_str;
    }
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

void SetOrderStatus(const std::string& status, ProductOrderElement* order) {
  if (status == "PENDING_CANCEL" || status == "REJECTED") {
    order->set_is_live(false);
  } else if (status == "NEW" || status == "CONFIRMED" || status == "PARTIALLY_FILLED") {
    order->set_is_live(true);
  } else if (status == "CANCELED" || status == "EXPIRED") {
    order->set_cancel_confirmed(true);
  } else if (status == "FILLED") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
}

bool IsBinanceFuturesErrMessage(const JsonValue& doc) {
  return doc.HasMember("code") && doc.HasMember("msg");
}

bool IsSuccessDataRequestResp(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  /*
  {
    "id": "12",
    "result": [
      {
        "req": "pxO5F3BH73G3BoZLuMbN2lLv6nLzHsDPchJwhyVy5H4R6iD4KGjUnmOakpQgJ84u@account",
        "res": {
          "feeTier": "9",
          "canTrade": true,
          "canDeposit": true,
          "canWithdraw": true,
          "accountAlias": "SgTiSgSgmYSguX"
        }
      }
    ]
  }
  */
  SPDLOG_INFO("[BinanceFutures] recv: {}", JsonToString(json));
  if (!json.HasMember("id") || !json.HasMember("result")) {
    return false;
  }
  if (!json["result"].IsArray()) {
    return false;
  }
  bool ret = false;
  for (const auto& data : json["result"].GetArray()) {
    if (!data.HasMember("req") || !data.HasMember("res")) {
      continue;
    }
    ret = true;
    break;
  }
  return ret;
}

bool IsAccountUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") && std::string_view(json["e"].GetString()) == "ACCOUNT_UPDATE";
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  /*
  {
    "e": "ORDER_TRADE_UPDATE",
    "T": 1589877664055,
    "E": 1589877664058,
    "o": {
      "s": "EOSUSDT",
      "c": "1589877658451818153",
      "S": "SELL",
      "o": "LIMIT",
      "f": "GTC",
      "q": "1",
      "p": "2.657",
      "ap": "0",
      "sp": "0",
      "x": "NEW",
      "X": "NEW",
      "i": 479299584,
      "l": "0",
      "z": "0",
      "L": "0",
      "T": 1589877664055,
      "t": 0,
      "b": "0",
      "a": "2.6570",
      "m": false,
      "R": false,
      "wt": "CONTRACT_PRICE",
      "ot": "LIMIT",
      "ps": "BOTH",
      "cp": false
    }
  }
  */
  return json.HasMember("e") && std::string_view(json["e"].GetString()) == "ORDER_TRADE_UPDATE";
}

}  // namespace

std::map<std::string, TransferType> BinanceFuturesParser::transfer_type_map_{
    {"1", TransferRequestProto::WITHDRAW_SPOT},
    {"2", TransferRequestProto::DEPOSIT_SPOT},
    {"3", TransferRequestProto::WITHDRAW_SPOT},
    {"4", TransferRequestProto::DEPOSIT_SPOT}};

BinanceFuturesParser::BinanceFuturesParser() {}

void BinanceFuturesParser::ParseRestAccountCrossMargin() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountCrossBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (!document.HasMember("userAssets")) {
    LOG(ERROR) << "[ParseRestAccountCrossBalance] unexpected format. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  for (const auto& asset : document["userAssets"].GetArray()) {
    const std::string native_currency = asset["asset"].GetString();
    if (!symcache_->has_currency_native(native_currency)) {
      continue;
    }
    account_cross_margin_cache_[native_currency] = stod(asset["netAsset"].GetString());
  }
  result_.set_type(MsgType::IGNORABLE);
  is_account_cross_margin_cached_ = true;
}

void BinanceFuturesParser::ParseRestAccountBalanceAndAccountPosition() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[" << __func__ << "] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (!document.HasMember("positions")) {
    LOG(ERROR) << "[" << __func__ << "] Don't has member \"positions\":" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (!document["positions"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Invalid Json:" << JsonToString(document["positions"]);
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_position = ParsePosition(document["positions"]);
  if (account_position.has_exchange() && account_position.has_market_type()) {
    result_.set_type(MsgType::ACCOUNT_POSITION);
    *result_.mutable_account_position() = std::move(account_position);
  }

  if (symcache_->IsPortfolioMarginTrading() && !is_account_cross_margin_cached_) {
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

bool BinanceFuturesParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  if (doc.IsObject() && IsBinanceFuturesErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.HasMember("assets") || !doc["assets"].IsArray()) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["assets"].GetArray()) {
    if (!data.HasMember("initialMargin") || !data.HasMember("marginBalance")) {
      continue;
    }
    double hold = stod(data["initialMargin"].GetString());
    double total = stod(data["marginBalance"].GetString());
    double avail = total - hold;
    CurrencyBalance balance;
    balance.set_currency_native(data["asset"].GetString());
    balance.set_available(avail);
    balance.set_hold(hold);
    balance.set_total(total);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance BinanceFuturesParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Binance");
  acc_balance.set_market_type("Futures");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    if (account_cross_margin_cache_.count(native_currency)) {
      const auto margin = account_cross_margin_cache_.at(native_currency);
      balance->set_total(balance->total() + margin);
      balance->set_available(balance->available() + margin);
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

int64_t BinanceFuturesParser::ParseRawPosition(
    const JsonValue& doc,
    RawPositionMap* raw_position_map) {
  int64_t position_update_et = -1;

  for (const auto& data : doc.GetArray()) {
    if (!data.HasMember("symbol") || !data.HasMember("positionAmt")) {
      LOG(ERROR) << "Unexpected position data! " << JsonToString(data);
      continue;
    }
    auto native_symbol = data["symbol"].GetString();
    auto& position = (*raw_position_map)[native_symbol];
    double pos_amt = stod(data["positionAmt"].GetString());
    if (pos_amt >= 0.) {
      position.set_long_position(pos_amt);
    } else {
      position.set_short_position(-pos_amt);
    }
    position.set_net_position(pos_amt);

    // let's treat them as option fields
    if (data.HasMember("initialMargin")) {
      double initial_margin = stod(data["initialMargin"].GetString());
      position.set_initial_margin(initial_margin);
    }
    if (data.HasMember("maintMargin")) {
      double maint_margin = stod(data["maintMargin"].GetString());
      position.set_maintenance_margin(maint_margin);
    }
    if (data.HasMember("leverage")) {
      double leverage = stod(data["leverage"].GetString());
      position.set_leverage(leverage);
    }
    if (data.HasMember("updateTime")) {
      position_update_et = std::max(position_update_et, stoll(data["updateTime"].GetString()));
    }
  }

  return position_update_et;
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountPosition BinanceFuturesParser::ParsePosition(const JsonValue& doc) {
  RawPositionMap raw_position_map;
  int64_t position_update_et = ParseRawPosition(doc, &raw_position_map);
  if (position_update_et + 1000 < position_newest_update_et_) {
    LOG(ERROR) << "delayed position rest query response fetch! position_newest_update_et:"
               << position_newest_update_et_ << " position_update_et:" << position_update_et;
    return {};
  }
  position_newest_update_et_ = std::max(position_newest_update_et_, position_update_et);

  AccountPosition acc_position;
  acc_position.set_exchange("Binance");
  acc_position.set_market_type("Futures");
  for (const auto& native_symbol : symcache_->GetProductNative()) {
    if (raw_position_map.count(native_symbol) != 0) {
      continue;
    }
    auto* position = acc_position.add_each_position();
    position->set_long_position(0);
    position->set_short_position(0);
    position->set_net_position(0);
    position->set_available_long_position(0);
    position->set_available_short_position(0);
    position->set_native_symbol(native_symbol);

    DLOG(INFO) << "POS: " << native_symbol << " " << position->net_position() << " "
               << position->initial_margin() << " " << position->maintenance_margin() << " "
               << position->leverage();
  }

  for (auto& pair : raw_position_map) {
    auto* position = acc_position.add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position() << " "
               << position->initial_margin() << " " << position->maintenance_margin() << " "
               << position->leverage();
  }
  is_account_position_initialized_ = true;

  return acc_position;
}

bool BinanceFuturesParser::ParseRawPositionFromWs(
    const JsonValue& doc,
    RawPositionMap* raw_position_map) {
  /*
  {
    "e": "ACCOUNT_UPDATE",
    "T": 1589879608104,
    "E": 1589879608107,
    "a": {
      "B": [
        {
          "a": "USDT",
          "wb": "92.42093499",
          "cw": "92.42093499"
        },
        {
          "a": "BNB",
          "wb": "0",
          "cw": "0"
        }
      ],
      "P": [
        {
          "s": "BTCUSDT",
          "pa": "-0.006",
          "ep": "9572.91000",
          "cr": "0.00063000",
          "up": "-1.21993",
          "mt": "cross",
          "iw": "0",
          "ps": "BOTH"
        },
        {
          "s": "EOSUSDT",
          "pa": "173",
          "ep": "2.6878",
          "cr": "2.01294041",
          "up": "-6.4010",
          "mt": "cross",
          "iw": "0",
          "ps": "BOTH"
        }
      ]
    }
  }
  */

  /*
I0602 14:48:53.829980 17631 parser.cc:209]
{"e":"ACCOUNT_UPDATE","T":"1591109333798","E":"1591109333803","a":{"B":[{"a":"USDT","wb":"64718.84580491","cw":"64718.84580491"},{"a":"BNB","wb":"132.54527273","cw":"0"}],"P":[{"s":"BTCUSDT","pa":"92.376","ep":"9711.65803","cr":"19749.88057328","up":"-2906.68751","mt":"cross","iw":"0","ps":"BOTH"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"LONG"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"SHORT"}]}}
I0602 14:48:53.844522 17631 parser.cc:209]
{"e":"ACCOUNT_UPDATE","T":"1591109333811","E":"1591109333816","a":{"B":[{"a":"USDT","wb":"64718.84580491","cw":"64718.84580491"},{"a":"BNB","wb":"132.51060431","cw":"0"}],"P":[{"s":"BTCUSDT","pa":"92.692","ep":"9711.28760","cr":"19749.88057328","up":"-2882.29481","mt":"cross","iw":"0","ps":"BOTH"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"LONG"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"SHORT"}]}}
I0602 14:48:53.844687 17631 parser.cc:209]
{"e":"ACCOUNT_UPDATE","T":"1591109333811","E":"1591109333816","a":{"B":[{"a":"USDT","wb":"64718.84580491","cw":"64718.84580491"},{"a":"BNB","wb":"132.50193615","cw":"0"}],"P":[{"s":"BTCUSDT","pa":"92.771","ep":"9711.19639","cr":"19749.88057328","up":"-2876.28878","mt":"cross","iw":"0","ps":"BOTH"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"LONG"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"SHORT"}]}}
I0602 14:48:53.859066 17631 parser.cc:209]
{"e":"ACCOUNT_UPDATE","T":"1591109333827","E":"1591109333832","a":{"B":[{"a":"USDT","wb":"64718.84580491","cw":"64718.84580491"},{"a":"BNB","wb":"132.46682900","cw":"0"}],"P":[{"s":"BTCUSDT","pa":"93.091","ep":"9710.82436","cr":"19749.88057328","up":"-2851.57747","mt":"cross","iw":"0","ps":"BOTH"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"LONG"},{"s":"BTCUSDT","pa":"0","ep":"0.00000","cr":"0","up":"0","mt":"cross","iw":"0","ps":"SHORT"}]}}
  */
  // document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (!doc.HasMember("T") || !doc["T"].IsString()) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  int64_t position_update_et = stoll(doc["T"].GetString());
  if (position_update_et + 1000 < position_newest_update_et_) {
    LOG(ERROR) << "delayed position websocket push fetch! position_newest_update_et:"
               << position_newest_update_et_ << " position_update_et:" << position_update_et;
    return false;
  }
  position_newest_update_et_ = std::max(position_newest_update_et_, position_update_et);

  if (!(doc.HasMember("a") && doc["a"].IsObject())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  if (!(doc["a"].HasMember("P") && doc["a"]["P"].IsArray())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }

  const auto& position_list = doc["a"]["P"];
  for (const auto& product_pos : position_list.GetArray()) {
    if (std::string_view(product_pos["ps"].GetString()) != "BOTH") {
      LOG(ERROR) << "Ws position ps not BOTH: " << JsonToString(doc);
      continue;
    }
    auto native_symbol = product_pos["s"].GetString();
    auto& position = (*raw_position_map)[native_symbol];
    auto net_pos = stod(product_pos["pa"].GetString());
    position.set_net_position(net_pos);
    if (net_pos >= 0) {
      position.set_long_position(net_pos);
    } else {
      position.set_short_position(-net_pos);
    }
  }

  return true;
}

bool BinanceFuturesParser::ParseWsPosition(const JsonValue& doc, AccountPosition* acc_position) {
  acc_position->Clear();

  RawPositionMap raw_position_map;
  bool success = ParseRawPositionFromWs(doc, &raw_position_map);
  if (!success) {
    return false;
  }

  acc_position->set_exchange("Binance");
  acc_position->set_market_type("Futures");
  for (auto& pair : raw_position_map) {
    auto* position = acc_position->add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position();
  }

  return true;
}

ProductOrderElement BinanceFuturesParser::ParseRestOrderInfo(const JsonValue& data) {
  ProductOrderElement order;
  std::string native_symbol = data["symbol"].GetString();
  if (!symcache_->has_product_native(native_symbol)) {
    return order;
  }
  std::string exchange_id = fmt::format("{}{}", native_symbol, data["orderId"].GetString());
  int64_t order_timestamp_ft = msg_->timestamp;
  int64_t order_timestamp_et = stoll(data["time"].GetString());
  if (order_timestamp_ft / 1e9 - order_timestamp_et / 1e3 > 60 * 60) {
    auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
    if (oc != nullptr && !GetOcm()->IsInWorkingOrders(oc)) {
      // TODO(xguo): adjust query url not to list all previous orders. Otherwise too chatty.
      // LOG(INFO) << "Ignore binance long time order: " << exchange_id;
      return order;
    }
  }
  order.set_market_type(MarketType::Futures);
  order.set_exchange_type(ExchangeType::Binance);
  order.set_api_version("v1");
  order.set_exchange_order_id(exchange_id);
  order.set_symbol(native_symbol);
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["origQty"].GetString()));
  order.set_proc_order_id(GetProcOrderId(data));
  order.set_side(GetOrderSide(data["side"].GetString()));
  order.set_order_type(GetOrderType(data["type"].GetString()));
  order.set_timestamp(order_timestamp_ft);
  SetOrderStatus(data["status"].GetString(), &order);

  const int64_t et = stoll(data["updateTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);

  return order;
}

void BinanceFuturesParser::ParseRestOpenOrders() {
  is_query_historical_orders_ = false;
  ParseRestOrderInfoList();
}

void BinanceFuturesParser::ParseRestAllOrders() {
  is_query_historical_orders_ = true;
  ParseRestOrderInfoList();
}

void BinanceFuturesParser::ParseRestOrderInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOpenOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsBinanceFuturesErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] " << JsonToString(data);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto new_order = ParseRestOrderInfo(info);

    // Ignore open orders form QueryHistoricalOrder response
    // only trust open orders from QueryOpenOrders response.
    if (!(is_query_historical_orders_ && new_order.is_live()) && new_order.has_market_type()) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
      product_order_bundle_found = true;
    }
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle BinanceFuturesParser::ParseWsOrderInfoListV2(const JsonValue& order_data) {
  /*
  {
    "e": "ORDER_TRADE_UPDATE",
    "T": 1589877664055,
    "E": 1589877664058,
    "o": {
      "s": "EOSUSDT",
      "c": "1589877658451818153",
      "S": "SELL",
      "o": "LIMIT",
      "f": "GTC",
      "q": "1",
      "p": "2.657",
      "ap": "0",
      "sp": "0",
      "x": "NEW",
      "X": "NEW",
      "i": 479299584,
      "l": "0",
      "z": "0",
      "L": "0",
      "T": 1589877664055,
      "t": 0,
      "b": "0",
      "a": "2.6570",
      "m": false,
      "R": false,
      "wt": "CONTRACT_PRICE",
      "ot": "LIMIT",
      "ps": "BOTH",
      "cp": false
    }
  }
  */

  ProductOrderBundle product_order_bundle;
  if (!order_data.IsObject()) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_order_bundle;
  }

  if (!order_data.HasMember("o")) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_order_bundle;
  }

  const auto& data = order_data["o"];
  auto* order = product_order_bundle.add_each_order();
  order->set_market_type(MarketType::Futures);
  order->set_exchange_type(ExchangeType::Binance);
  order->set_api_version("v1");

  order->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
  order->set_symbol(data["s"].GetString());
  order->set_price(stod(data["p"].GetString()));
  order->set_qty(stod(data["q"].GetString()));
  order->set_proc_order_id(stoll(data["c"].GetString()));
  order->set_side(GetOrderSide(data["S"].GetString()));
  order->set_order_type(GetOrderType(data["o"].GetString()));
  SetOrderStatus(data["X"].GetString(), order);
  order->set_timestamp(msg_->timestamp);

  const int64_t et = stoll(data["T"].GetString()) * 1000'000LL;
  wss_transaction_update_et_ = std::max(wss_transaction_update_et_, et);

  return product_order_bundle;
}

ProductFillBundle BinanceFuturesParser::ParseWsFillInfoListV2(const JsonValue& order_data) {
  /*
  {
    "e": "ORDER_TRADE_UPDATE",
    "T": 1589879608104,
    "E": 1589879608107,
    "o": {
      "s": "EOSUSDT",
      "c": "1589879106495286828",
      "S": "SELL",
      "o": "LIMIT",
      "f": "GTC",
      "q": "1",
      "p": "2.654",
      "ap": "2.6540",
      "sp": "0",
      "x": "TRADE",
      "X": "FILLED",
      "i": 479397583,
      "l": "1",
      "z": "1",
      "L": "2.654",
      "n": "-0.00005308",
      "N": "USDT",
      "T": 1589879608104,
      "t": 10210519,
      "b": "0",
      "a": "0",
      "m": true,
      "R": false,
      "wt": "CONTRACT_PRICE",
      "ot": "LIMIT",
      "ps": "BOTH",
      "cp": false
    }
  }
  */
  ProductFillBundle product_fill_bundle;
  if (!order_data.HasMember("o")) {
    LOG(ERROR) << "Unexpected json format! " << JsonToString(order_data);
    return product_fill_bundle;
  }

  const auto& data = order_data["o"];
  std::string_view status = data["X"].GetString();
  if (status == "FILLED" || status == "PARTIALLY_FILLED" || status == "TRADE") {
    auto* fill = product_fill_bundle.add_each_fill();
    fill->set_symbol(data["s"].GetString());
    fill->set_fill_timestamp(msg_->timestamp);
    fill->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
    fill->set_proc_order_id(stoll(data["c"].GetString()));
    fill->set_price(stod(data["L"].GetString()));
    fill->set_qty(stod(data["l"].GetString()));
    fill->set_fill_id(data["t"].GetString());
    if (data["m"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (data.HasMember("N") && !data["N"].IsNull()) {
      const std::string fee_currency = data["N"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (data.HasMember("n") && !data["n"].IsNull()) {
          const std::string fee = data["n"].GetString();
          if (!fee.empty()) {
            fill->set_fee(stod(fee));
          }
        }
      }
    }
  }

  const int64_t et = stoll(data["T"].GetString()) * 1000'000LL;
  wss_transaction_update_et_ = std::max(wss_transaction_update_et_, et);

  return product_fill_bundle;
}

void BinanceFuturesParser::ParseRestFillInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected rest fill msg!" << JsonToString(data);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(info["symbol"].GetString());
    fill->set_fill_timestamp(msg_->timestamp);
    fill->set_exchange_order_id(
        info["symbol"].GetString() + std::string(info["orderId"].GetString()));
    fill->set_price(stod(info["price"].GetString()));
    fill->set_qty(stod(info["qty"].GetString()));
    fill->set_fill_id(info["id"].GetString());
    if (info["maker"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (info.HasMember("commissionAsset") && !info["commissionAsset"].IsNull()) {
      const std::string fee_currency = info["commissionAsset"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (info.HasMember("commission") && !info["commission"].IsNull()) {
          const std::string fee = info["commission"].GetString();
          if (!fee.empty()) {
            fill->set_fee(stod(fee));
          }
        }
      }
    }

    const int64_t et = stoll(info["time"].GetString()) * 1'000'000LL;
    rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);

    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

std::string BinanceFuturesParser::ParseHeader(
    const std::shared_ptr<AsioHttpContext>& http_context) {
  const auto& header = http_context->res.base();
  nlohmann::json json = {
      {"x-mbx-order-count-1d", -1},
      {"x-mbx-order-count-10s", -1},
      {"x-mbx-order-count-1m", -1},
      {"x-mbx-used-weight-1m", -1},
      {"retry-after", -1},
      {"bind_ip", http_context->bind_ip}};

  if (header.count("x-mbx-used-weight-1m") > 0) {
    json["x-mbx-used-weight-1m"] = std::stoi(header.at("x-mbx-used-weight-1m").data());
  }

  if (header.count("x-mbx-order-count-1m") > 0) {
    json["x-mbx-order-count-1m"] = std::stoi(header.at("x-mbx-order-count-1m").data());
  }

  if (header.count("x-mbx-order-count-10s") > 0) {
    json["x-mbx-order-count-10s"] = std::stoi(header.at("x-mbx-order-count-10s").data());
  }

  if (header.count("retry-after") > 0) {
    json["retry-after"] = std::stoi(header.at("retry-after").data());
  } else if (http_context->res.result_int() == 403) {
    LOG(ERROR)
        << "WAF limit hit! block request for 6 minutes prevent WAF block time extend longer.";
    json["retry-after"] = 6 * 60;
  }
  return json.dump();
}

bool BinanceFuturesParser::IsValidBinanceResponse(
    const std::shared_ptr<AsioHttpContext>& http_context) {
  /*
  http_response {
    timestamp: 1679641708204068428
    status_code: 503
    headers {
      key: "Connection"
      value: "keep-alive"
    }
    headers {
      key: "Content-Length"
      value: "963"
    }
    headers {
      key: "Content-Type"
      value: "text/html"
    }
    headers {
      key: "Date"
      value: "Fri, 24 Mar 2023 07:08:28 GMT"
    }
    headers {
      key: "Server"
      value: "CloudFront"
    }
    headers {
      key: "Via"
      value: "1.1 6259584a0d6b38c5f298323a776f2b74.cloudfront.net (CloudFront)"
    }
    headers {
      key: "X-Amz-Cf-Id"
      value: "EYnZG6f6l5NWBG9FgxFhbIbLqESLEO7CLoW-0AyKMcHTE0Xn0kjf7A=="
    }
    headers {
      key: "X-Amz-Cf-Pop"
      value: "NRT20-C1"
    }
    headers {
      key: "X-Cache"
      value: "LimitExceeded from cloudfront"
    }
    body: "<!DOCTYPE HTML PUBLIC \"-//W3C//DTD HTML 4.01 Transitional//EN\"
  \"http://www.w3.org/TR/html4/loose.dtd\">\n<HTML><HEAD><META HTTP-EQUIV=\"Content-Type\"
  CONTENT=\"text/html; charset=iso-8859-1\">\n<TITLE>ERROR: The request could not be
  satisfied</TITLE>\n</HEAD><BODY>\n<H1>503 ERROR</H1>\n<H2>The request could not be
  satisfied.</H2>\n<HR noshade size=\"1px\">\nThis distribution exceeded the limit on requests per
  second.\nWe can\'t connect to the server for this app or website at this time. There might be too
  much traffic or a configuration error. Try again later, or contact the app or website owner.\n<BR
  clear=\"all\">\nIf you provide content to customers through CloudFront, you can find steps to
  troubleshoot and help prevent this error by reviewing the CloudFront documentation.\n<BR
  clear=\"all\">\n<HR noshade size=\"1px\">\n<PRE>\nGenerated by cloudfront (CloudFront)\nRequest
  ID:
  EYnZG6f6l5NWBG9FgxFhbIbLqESLEO7CLoW-0AyKMcHTE0Xn0kjf7A==\n</PRE>\n<ADDRESS>\n</ADDRESS>\n</BODY></HTML>"
  }
  */

  // special handle 503 code with cloudfront
  if (http_context->res.result_int() == 503) {
    std::string body(http_context->res.body().data(), http_context->res.body().length());

    if (body.find("Generated by cloudfront") != std::string::npos ||
        body.find(R"("code":-1001,)") != std::string::npos ||
        body.find("Service Unavailable.") != std::string::npos) {
      return true;
    }
  }

  return IsValidResponse(http_context);
}

void BinanceFuturesParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.base().result_int());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BinanceFuturesSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[BinanceFuturesSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BinanceFuturesSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!json.HasMember("orderId")) {
    LOG(ERROR) << "[BinanceFutures Submit Order] Error Response" << http_context->res;
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(
      json["symbol"].GetString() + std::string(json["orderId"].GetString()));

  const int64_t et = stoll(json["updateTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
}

void BinanceFuturesParser::ParseBatchSubmitResponse() {
  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_SUBMIT_RESPONSE);
  auto* response = result_.mutable_multi_submit_response();
  response->Clear();

  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_submit_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.base().result_int());
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json) || !json.IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  /*
    match order_context and response in order due to binance api docs said as below.

    https://binance-docs.github.io/apidocs/futures/en/#place-multiple-orders-trade
    ```
      The order of returned contents for batch orders is the same as the order of the order list.
    ```

    and if one order in order list submit fail, binance will return response like below
    which not have a client order id to match with order_context.
    only can rely on in this situation is the order of response and the order of order_context.

    [
        {
            "clientOrderId":"testOrder",
            "cumQty":"0",
            "cumQuote":"0",
            "executedQty":"0",
            "orderId":22542179,
            "avgPrice":"0.00000",
            "origQty":"10",
            "price":"0",
            "reduceOnly":false,
            "side":"BUY",
            "positionSide":"SHORT",
            "status":"NEW",
            "stopPrice":"9300",
            "symbol":"BTCUSDT",
            "timeInForce":"GTC",
            "type":"TRAILING_STOP_MARKET",
            "origType":"TRAILING_STOP_MARKET",
            "activatePrice":"9020",
            "priceRate":"0.3",
            "updateTime":1566818724722,
            "workingType":"CONTRACT_PRICE",
            "priceProtect":false
        },
        {
            "code":-2022,
            "msg":"ReduceOnly Order is rejected."
        }
    ]
  */

  CHECK_EQ(response->each_submit_response_size(), json.GetArray().Size())
      << "Batch submit response size not equal to request size response:" << JsonToString(json)
      << " request:" << response->DebugString();
  for (int i = 0; i < json.GetArray().Size(); ++i) {
    const auto& data = json.GetArray()[i];
    auto* resp = response->mutable_each_submit_response(i);

    if (!data.HasMember("orderId")) {
      LOG(ERROR) << "[" << __func__ << "] Error Response" << JsonToString(data);
      auto error_code = GetOrderErrorCode(data["code"], data["msg"]);
      resp->set_error_code(error_code);
      resp->set_exchange_error_code(GetExchangeOrderErrorCode(data));
      resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(data));
      continue;
    }

    CHECK_EQ(data["clientOrderId"].GetString(), std::to_string(resp->proc_order_id()))
        << "client order id different response:" << data["clientOrderId"].GetString()
        << " expect:" << resp->proc_order_id();

    resp->set_success(true);
    resp->set_exchange_order_id(
        data["symbol"].GetString() + std::string(data["orderId"].GetString()));

    const int64_t et = stoll(data["updateTime"].GetString()) * 1'000'000LL;
    rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
  }
}

void BinanceFuturesParser::ParseCancelResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.base().result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BinanceFuturesCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[BinanceFuturesCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BinanceFuturesCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("symbol") && json.HasMember("orderId")) {
    resp->set_exchange_order_id(
        json["symbol"].GetString() + std::string(json["orderId"].GetString()));
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[BinanceFuturesCancelOrder] Error Response" << http_context->res;
      resp->set_success(false);
      return;
    }
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);

  const int64_t et = stoll(json["updateTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
}

void BinanceFuturesParser::ParseBatchCancelResponse() {
  /*
  [
      {
          "orderId":8389765609835744000,
          "symbol":"ETHUSDT",
          "status":"CANCELED",
          "clientOrderId":"1690191474708259119",
          "price":"1866.56",
          "avgPrice":"0.00",
          "origQty":"0.004",
          "executedQty":"0.000",
          "cumQty":"0.000",
          "cumQuote":"0.00000",
          "timeInForce":"GTX",
          "type":"LIMIT",
          "reduceOnly":false,
          "closePosition":false,
          "side":"BUY",
          "positionSide":"BOTH",
          "stopPrice":"0.00",
          "workingType":"CONTRACT_PRICE",
          "priceProtect":false,
          "origType":"LIMIT",
          "updateTime":1690191512315
      },
      {
          "code":-2011,
          "msg":"Unknown order sent."
      }
  ]
  */

  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_CANCEL_RESPONSE);
  auto* response = result_.mutable_multi_cancel_response();
  response->Clear();

  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_cancel_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.base().result_int());
    resp->set_exchange_order_id(order_context->exchange_order_id());
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json) || !json.IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  const auto& array = json.GetArray();

  CHECK_EQ(response->each_cancel_response_size(), array.Size())
      << "Batch cancel response size not equal to request size msg:" << msg_->DebugString();
  for (int i = 0; i < array.Size(); ++i) {
    const auto& data = array[i];
    auto* resp = response->mutable_each_cancel_response(i);

    if (data.HasMember("code") && data.HasMember("msg")) {
      auto error_code = GetOrderErrorCode(data["code"], data["msg"]);
      resp->set_error_code(error_code);
      resp->set_exchange_error_code(GetExchangeOrderErrorCode(data));
      resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(data));
      if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
        LOG(ERROR) << "[" << __func__ << "] Error Response" << JsonToString(data);
        continue;
      }
    }

    CHECK_EQ(data["clientOrderId"].GetString(), std::to_string(resp->proc_order_id()))
        << "client order id different response:" << data["clientOrderId"].GetString()
        << " expect:" << resp->proc_order_id();

    resp->set_success(true);
    resp->set_cancel_confirmed(true);
    resp->set_exchange_order_id(
        data["symbol"].GetString() + std::string(data["orderId"].GetString()));

    const int64_t et = stoll(data["updateTime"].GetString()) * 1'000'000LL;
    rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
  }
}

void BinanceFuturesParser::ParseTransfer() {
  /*
  {
    "rows":
    [
      {
        "asset": "USDT",
        "tranId": 100000001
        "amount": "40.84624400",
        "type": "1",
        "timestamp": 1555056425000,
        "status": "CONFIRMED"
      }
    ],
    "total": 1
  }
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    return;
  }

  if (data.IsObject() && IsBinanceFuturesErrMessage(data)) {
    LOG(ERROR) << "[ParseTransfer] " << JsonToString(data);
    return;
  }

  if (std::string_view(data["total"].GetString()) == "0") {
    LOG(INFO) << "[ParseTransfer] no transfer: " << JsonToString(data);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& row : data["rows"].GetArray()) {
    if (!row.HasMember("asset") || !row.HasMember("type") || !row.HasMember("amount")) {
      LOG(ERROR) << "[ParseTransfer] Invalid transfer format";
      continue;
    }
    if (!symcache_->has_currency_native(row["asset"].GetString())) {
      continue;
    }
    auto* currency = symcache_->GetCurrencyByNative(row["asset"].GetString());
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    std::string type = row["type"].GetString();
    if (transfer_type_map_.find(type) != transfer_type_map_.end()) {
      currency_transfer->set_type(transfer_type_map_[type]);
    }
    std::string status = row["status"].GetString();
    if (status == "CONFIRMED") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_transfer_id(row["tranId"].GetString());
    currency_transfer->set_qty(stod(row["amount"].GetString()));
    int64_t ts = std::stoll(row["timestamp"].GetString()) * 1e6;
    currency_transfer->set_create_timestamp(ts);
    if (ts > result_.last_transfer_timestamp_) {
      result_.last_transfer_timestamp_ = ts;
    }
    currency_transfer_bundle_found = true;
  }

  if (currency_transfer_bundle_found) {
    result_.set_type(BaseParser::MsgType::TRANSFER);
  }
}

void BinanceFuturesParser::CheckAndForceResetWssConnection() {
  if (wss_transaction_update_et_ == 0 ||
      wss_transaction_update_et_ + 300'000'000'000LL /* 5 mins */ > rest_transaction_update_et_) {
    return;
  }
  mb_->PublishEvent("force_reset_wss_connection", &wss_transaction_update_et_ /* dummy */);
}

void BinanceFuturesParser::ParseFundingFee() {
  /*
  [
      {
          "symbol":"EOSUSDT",
          "incomeType":"FUNDING_FEE",
          "income":"-0.00006608",
          "asset":"USDT",
          "time":1691164801000,
          "info":"FUNDING_FEE",
          "tranId":7398752307368348460,
          "tradeId":""
      },
      {
          "symbol":"TRXUSDT",
          "incomeType":"FUNDING_FEE",
          "income":"-0.00087801",
          "asset":"USDT",
          "time":1691164801000,
          "info":"FUNDING_FEE",
          "tranId":7398752315840840370,
          "tradeId":""
      }
  ]
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    return;
  }

  if (data.IsObject() && IsBinanceFuturesErrMessage(data)) {
    LOG(ERROR) << "[" << __func__ << "] " << JsonToString(data);
    return;
  }

  if (!data.IsArray()) {
    LOG(INFO) << "[" << __func__ << "] Parse error: " << JsonToString(data);
    return;
  }

  std::vector<FundingFeeElement> funding_fees;
  for (const auto& record : data.GetArray()) {
    if (!record.HasMember("symbol") || !record.HasMember("income") || !record.HasMember("asset") ||
        !record.HasMember("time") || !record.HasMember("tranId")) {
      LOG(ERROR) << "[" << __func__ << "] Invalid funding fee format:" << JsonToString(record);
      continue;
    }

    const std::string asset = record["asset"].GetString();
    const std::string symbol = record["symbol"].GetString();
    if (!symcache_->has_currency_native(asset) || !symcache_->has_product_native(symbol)) {
      continue;
    }

    auto* currency = symcache_->GetCurrencyByNative(asset);
    auto* product = symcache_->GetProductByNative(symbol);
    FundingFeeElement funding_fee;

    funding_fee.set_id(record["tranId"].GetString());
    funding_fee.set_symbol(product->symbol());
    funding_fee.set_funding_fee(-stod(record["income"].GetString()));
    funding_fee.set_funding_fee_currency(currency->currency());
    funding_fee.set_timestamp(stoll(record["time"].GetString()) * 1'000'000);

    funding_fees.emplace_back(std::move(funding_fee));
  }

  if (is_first_funding_fee_) {
    is_first_funding_fee_ = false;

    for (const auto& funding_fee : funding_fees) {
      consumed_funding_fee_ids_.emplace(funding_fee.id());
    }
    return;
  }

  auto funding_fee_bundle = result_.mutable_funding_fee_bundle();
  funding_fee_bundle->Clear();
  bool funding_fee_bundle_found = false;
  for (auto& funding_fee : funding_fees) {
    if (consumed_funding_fee_ids_.count(funding_fee.id()) != 0) {
      continue;
    }
    consumed_funding_fee_ids_.emplace(funding_fee.id());

    *funding_fee_bundle->add_each_funding_fee() = std::move(funding_fee);
    funding_fee_bundle_found = true;
  }

  if (funding_fee_bundle_found) {
    result_.set_type(BaseParser::MsgType::FUNDING_FEE);
  }
}

void BinanceFuturesParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  bool update_ws_last_recv_ts = false;
  if (IsAccountUpdate(document)) {
    // make first rest position result before ws position result,
    // so SetAllZeroPosition won't take effect on partial symbol
    if (!is_account_position_initialized_) {
      result_.set_type(MsgType::IGNORABLE);
    } else {
      if (ParseWsPosition(document, result_.mutable_account_position())) {
        result_.set_type(MsgType::ACCOUNT_POSITION);
      } else {
        result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      }
      // linchuan : ws balance lack of all symbol's unrealized profit, disabled.
      // result_.set_type(MsgType::ACCOUNT_BALANCE);
    }
    update_ws_last_recv_ts = true;
  } else if (IsOrderUpdate(document)) {
    *result_.mutable_fill_list() = ParseWsFillInfoListV2(document);
    *result_.mutable_order_list() = ParseWsOrderInfoListV2(document);
    result_.set_type(MsgType::ORDER_STATUS);
    result_.set_type(MsgType::ORDER_FILL);
    update_ws_last_recv_ts = true;
  } else if (IsSuccessDataRequestResp(document)) {
    result_.set_type(MsgType::IGNORABLE);
    update_ws_last_recv_ts = true;
  }

  if (update_ws_last_recv_ts) {
    int64_t timestamp = msg_->timestamp;
    mb_->PublishEvent("upd_ws_last_recv_msg_ts", &timestamp);
  }
}

void BinanceFuturesParser::ParseListenKey() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseListenKey] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // {"listenKey":"fmt5gmPtRPVO6QOf1bP2euju10VoZ0xCvXmh6f3LrvGEyeuh3tgOh3liN06O"}
  LOG(INFO) << "binance/futures: " << JsonToString(document);
  if (document.HasMember("listenKey")) {
    std::string listen_key(document["listenKey"].GetString());
    mb_->PublishEvent("binance/listen_key", &listen_key);
    last_key_query_ts_ = msg_->timestamp;
  } else {
    double delta_ts = (msg_->timestamp - last_key_query_ts_) / 1e9;
    LOG(ERROR) << "Failed to query listenKey" << JsonToString(document);
    LOG(INFO) << "key_age(seconds): " << delta_ts;
  }
}

auto BinanceFuturesParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg_->type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  std::string s = ParseHeader(msg.http_context);
  mb_->PublishEvent("binance/req_rate_limit", reinterpret_cast<void*>(&s));

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::LISTEN_KEY:
      ParseListenKey();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      ParseBatchCancelResponse();
      break;
    case OrderSubTopicId::MULTI_SUBMIT_RESPONSE:
      ParseBatchSubmitResponse();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestAllOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalanceAndAccountPosition();
      break;
    case OrderSubTopicId::REST_ACCOUNT_CROSS_BALANCE:
      ParseRestAccountCrossMargin();
      break;
    case OrderSubTopicId::TRANSFER:
      ParseTransfer();
      break;
    case OrderSubTopicId::REST_FUNDING_FEE:
      ParseFundingFee();
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  CheckAndForceResetWssConnection();

  return result_;
}

}  // namespace coin2::exchange::binance_futures::order_v1
